#pragma once
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/MqProto.pb.h"
#include "../mqcommon/ThreadPool.hpp"
#include "VirtualHost.hpp"
#include "Consumer.hpp"
#include "Route.hpp"

namespace ouyangmq
{
    // 12 个调用请求使用智能指针封装
    // 信道打开和关闭
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;

    // 交换机的声明和删除
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;

    // 队列的声明和删除
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;

    // 队列的绑定和解绑
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;

    // 订阅和取消订阅
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    // 发布和确认
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;

    // 使用智能指针封装protobuf协议处理器
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &id, const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp,
                const VirtualHost::ptr &vhp, const threadpool::ptr &pool)
            : _cid(id), _conn(conn), _codec(codec), _cmp(cmp), _vhp(vhp), _pool(pool)
        {
            LogModule::LOG(LogModule::LogLevel::INFO) << "new Channel" << this;
        }

        ~Channel()
        {
            // 信道关闭，移除消费者
            if (_consumer.get() != nullptr)
            {
                _cmp->RemoveConsumer(_consumer->tag, _consumer->qname);
            }
        }

        // 1. 提供声明&删除交换机操作
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            bool ret = _vhp->declareExchange(req->exchange_name(), req->exchange_type(), req->durable(),
                                             req->auto_delete(), req->args());
            return basicResponse(ret, req->rid(), req->cid());
        }
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            _vhp->deleteExchange(req->exchange_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 2. 提供声明&删除队列操作
        void declareQueue(const declareQueueRequestPtr &req)
        {
            bool ret = _vhp->declareQueue(req->queue_name(), req->durable(), req->exclusive(),
                                          req->auto_delete(), req->args());
            if (ret == false)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            // 初始化消费者队列
            _cmp->InitQueueConsumer(req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        void deleteQueue(const deleteQueueRequestPtr &req)
        {
            // 删除消费者队列
            _cmp->DestoryQueueConsumer(req->queue_name());
            _vhp->deleteQueue(req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 3. 提供绑定&解绑队列操作
        void queueBind(const queueBindRequestPtr &req)
        {
            bool ret = _vhp->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            return basicResponse(ret, req->rid(), req->cid());
        }
        void queueUnBind(const queueUnBindRequestPtr &req)
        {
            _vhp->unbind(req->exchange_name(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 4. 发布&确认消息操作
        void basicPublish(const basicPublishRequestPtr &req)
        {
            // 1. 找到指定的交换机, 查看交换机是否存在
            auto ep = _vhp->GetOneExchage(req->exchange_name());
            if (ep.get() == nullptr)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "找不到指定的交换机进行发布消息" << req->exchange_name();
                return basicResponse(false, req->rid(), req->cid());
            }
            // 2. 获取指定交换机的所有绑定信息
            MsgQueueBindingMap exchange_bindings = _vhp->exchangeBindings(req->exchange_name());
            // 3. 通过交换路由, 将信息发布到指定的队列中去
            // [1] 拿到req中的routing_key, 如果有就拿取
            std::string routing_key;
            BasicProperties *properties = nullptr;
            if (req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }
            for (auto &bg : exchange_bindings)
            {
                if (Router::route(ep->type, routing_key, bg.second->binding_key))
                {
                    // 4. 将消息推送到指定的队列之中(消息管理)
                    _vhp->basicPublish(bg.first, properties, req->body());
                    // 5. 向线程池中添加consume的任务
                    auto task = std::bind(&Channel::consume, this, bg.first);
                    _pool->Push(task);
                }
            }
            return basicResponse(true, req->rid(), req->cid());
        }

        // 5. 提供订阅&取消订阅队列消息操作
        void basicConsumer(const basicConsumeRequestPtr &req)
        {
            // 1. 判断队列是否存在
            bool ret = _cmp->exists(req->consumer_tag(), req->queue_name());
            if (ret == false)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            // 2. 创建队列的消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1,
                                std::placeholders::_2, std::placeholders::_3);
            // 将消息推送给自己this
            _cmp->AddConsumer(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            return basicResponse(true, req->rid(), req->cid());
        }

        void basicCancel(const basicCancelRequestPtr &req)
        {
            _cmp->RemoveConsumer(req->consumer_tag(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 6. 消息确认
        void basicAck(const basicAckRequestPtr &req)
        {
            _vhp->basicAck(req->queue_name(), req->message_id());
            return basicResponse(true, req->rid(), req->cid());
        }

    private:
        // 回调函数
        void callback(const std::string &tag, const ouyangmq::BasicProperties *bp, const std::string &body)
        {
            // 针对参数组织出推送消息请求，将消息推送给channel对应的客户端
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }

        // 消费消息
        void consume(const std::string &qname)
        {
            // 1. 从队列中取出一条消息消费
            MessagePtr msp = _vhp->basicConsume(qname);
            if (msp.get() == nullptr)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "执行消费消息失败, 该队列中没有消息 " << qname;
                return;
            }
            // 2. 从队列订阅者中取出一个订阅者
            Consumer::ptr cp = _cmp->Choose(qname);
            if (cp.get() == nullptr)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "执行消费消息失败, 该队列没有订阅者 " << qname;
                return;
            }
            // 3. 调用订阅者对应的回调函数
            cp->callbackfunc(cp->tag, msp->mutable_payload()->mutable_properties(), msp->payload().body());
            // 4. 判定该订阅者是否为自动确认, 如果是自动确认ack直接删除消息
            if (cp->auto_ack)
                _vhp->basicAck(qname, msp->payload().properties().id());
        }

        // 响应函数
        void basicResponse(bool ok, const std::string &rid, const std::string &cid)
        {
            // 响应
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            // 发送
            _codec->send(_conn, resp);
        }

    private:
        // 1. 信道的ID, 唯一标识
        std::string _cid;
        // 2. 信道关联的消费者
        Consumer::ptr _consumer;
        // 3. 信道关联的连接
        muduo::net::TcpConnectionPtr _conn;
        // 4. protobuf协议处理句柄
        ProtobufCodecPtr _codec;
        // 5. 消费者管理句柄
        ConsumerManager::ptr _cmp;
        // 6. 虚拟机句柄
        VirtualHost::ptr _vhp;
        // 7. 工作线程池
        threadpool::ptr _pool;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {};
        bool openChannel(const std::string &id, const muduo::net::TcpConnectionPtr &conn,
                         const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp,
                         const VirtualHost::ptr &vhp, const threadpool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 如果在_channels可以找到说明已经存在了, 不需要再次创建
            auto it = _channels.find(id);
            if (it != _channels.end())
                return false;
            // 没有存在就需要创建
            Channel::ptr cp = std::make_shared<Channel>(id, conn, codec, cmp, vhp, pool);
            _channels.emplace(id, cp);
            return true;
        }

        void closeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }

        Channel::ptr GetOneChannel(const std::string &id)
        {
            Channel::ptr cp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 查看是否存在, 如果不存在返回一个空对象
                auto it = _channels.find(id);
                if (it == _channels.end())
                    return Channel::ptr();
                cp = it->second;
            }
            return cp;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}